package com.atguigu.gmall.realtime.dim.app;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.io.IOException;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.*;

public class DimApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.环境准备
        // 1.1 指定流处理环境
        Configuration conf = new Configuration();
        conf.set(RestOptions.PORT, 8088);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        // 1.2 设置并行度
        env.setParallelism(4);

        // TODO 2.检查点相关的设置
        // 2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 2.2 设置检查点超时时间
        checkpointConfig.setCheckpointTimeout(60000L);
        // 2.3 设置job取消后检查点是否保留
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.4 设置两个检查点之间时间间隔
        checkpointConfig.setMinPauseBetweenCheckpoints(2000L);
        // 2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
        // 2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        // 2.7 设置检查点存储路径
        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/ck/");
        // 2.8 设置操作Hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");

        //TODO 3.从kafka的主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String groupId = "dim_app_group";
        //3.2 创建KafkaSource消费者对象
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                .setTopics(Constant.TOPIC_DB)
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(
                        new DeserializationSchema<String>() {
                            @Override
                            public String deserialize(byte[] message) throws IOException {
                                if (message != null) {
                                    return new String(message);
                                }
                                return null;
                            }

                            @Override
                            public boolean isEndOfStream(String s) {
                                return false;
                            }

                            @Override
                            public TypeInformation<String> getProducedType() {
                                return TypeInformation.of(String.class);
                            }
                        }
                ).build();
        //3.3 获取kafkaSource 封装为流
        DataStreamSource<String> kafkaStrDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //TODO 4.对流中数据进行类型转换并进行简单的ETL    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        JSONObject jsonObj = JSONObject.parseObject(jsonStr);
                        String db = jsonObj.getString("database");
                        String type = jsonObj.getString("type");
                        String data = jsonObj.getString("data");
                        if ("gmall0221".equals(db)
                                && ("insert".equals(type)
                                || "update".equals(type)
                                || "delete".equals(type)
                                || "bootstrap-insert".equals(type))
                                && data != null
                                && data.length() > 2) {
                            collector.collect(jsonObj);
                        }
                    }
                }
        );
        //jsonObjDS.print();
        //TODO 5.使用FlinkCDC从配置表中读取配置信息
        //5.1 创建MySqlSource对象
        Properties props = new Properties();
        props.setProperty("useSSL", "false");
        props.setProperty("allowPublicKeyRetrieval", "true");
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("gmall0221_config")
                .tableList("gmall0221_config.table_process_dim")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .jdbcProperties(props)
                .build();
        //5.2 读取数据 封装为流
        DataStreamSource<String> mysqlStrDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql_source");
        //TODO 6.对flinkCDC读取到的数据进行类型转换  jsonStr->实体类对象
        SingleOutputStreamOperator<TableProcessDim> tpDS = mysqlStrDS.map(
                new MapFunction<String, TableProcessDim>() {
                    @Override
                    public TableProcessDim map(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSONObject.parseObject(jsonStr);
                        String op = jsonObj.getString("op");
                        TableProcessDim tableProcessDim = null;
                        if ("d".equals(op)) {
                            tableProcessDim = jsonObj.getObject("before", TableProcessDim.class);
                        } else {
                            tableProcessDim = jsonObj.getObject("after", TableProcessDim.class);
                        }
                        tableProcessDim.setOp(op);
                        return tableProcessDim;
                    }
                }
        ).setParallelism(1);
        //tpDS.print();
        //TODO 7.在广播配置信息前，根据配置流中的配置在HBase中执行建表或者删除表操作
        tpDS = tpDS.map(
                new RichMapFunction<TableProcessDim, TableProcessDim>() {
                    private Connection hBaseConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hBaseConn = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hBaseConn);
                    }

                    @Override
                    public TableProcessDim map(TableProcessDim tableProcessDim) throws Exception {
                        String op = tableProcessDim.getOp();
                        String sinkTable = tableProcessDim.getSinkTable();
                        String[] families = tableProcessDim.getSinkFamily().split(",");
                        if ("c".equals(op) || "r".equals(op)) {
                            HBaseUtil.createHBaseTable(hBaseConn, Constant.HBASE_NAMESPACE, sinkTable, families);
                        } else if ("d".equals(op)) {
                            HBaseUtil.dropHBaseTable(hBaseConn, Constant.HBASE_NAMESPACE, sinkTable);
                        } else {
                            HBaseUtil.createHBaseTable(hBaseConn, Constant.HBASE_NAMESPACE, sinkTable, families);
                            HBaseUtil.dropHBaseTable(hBaseConn, Constant.HBASE_NAMESPACE, sinkTable);
                        }
                        return tableProcessDim;
                    }
                }
        );
        //TODO 8.广播配置流---broadcast
        MapStateDescriptor<String, TableProcessDim> mapStateDescriptor
                = new MapStateDescriptor<String, TableProcessDim>("mapStateDescriptor", String.class, TableProcessDim.class);
        BroadcastStream<TableProcessDim> broadcastDS = tpDS.broadcast(mapStateDescriptor);
        //TODO 9.关联主流业务数据以及广播流配置信息---connect
        BroadcastConnectedStream<JSONObject, TableProcessDim> connectDS = jsonObjDS.connect(broadcastDS);
        //TODO 10.对关联后的数据进行处理---process
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> dimDS = connectDS.process(
                new BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>() {
                    private Map<String, TableProcessDim> configMap = new HashMap<>();

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        Class.forName(Constant.MYSQL_DRIVER);
                        java.sql.Connection conn = DriverManager.getConnection(Constant.MYSQL_URL, Constant.MYSQL_USER_NAME, Constant.MYSQL_PASSWORD);
                        String sql = "select * from gmall0221_config.table_process_dim";
                        PreparedStatement ps = conn.prepareStatement(sql);
                        ResultSet rs = ps.executeQuery();
                        ResultSetMetaData metaData = rs.getMetaData();
                        while (rs.next()) {
                            JSONObject jsonObj = new JSONObject();
                            for (int i = 1; i < metaData.getColumnCount(); i++) {
                                String columnName = metaData.getColumnName(i);
                                Object columnValue = rs.getObject(i);
                                jsonObj.put(columnName, columnValue);
                            }
                            TableProcessDim tableProcessDim = jsonObj.toJavaObject(TableProcessDim.class);
                            configMap.put(tableProcessDim.getSourceTable(), tableProcessDim);
                        }
                        rs.close();
                        ps.close();
                        conn.close();
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
                        String table = jsonObject.getString("table");
                        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
                        TableProcessDim tableProcessDim = null;
                        if ((tableProcessDim = broadcastState.get(table)) != null
                                || configMap.get(table) != null) {
                            JSONObject dataJsonObj = jsonObject.getJSONObject("data");
                            //tableProcessDim = Optional.ofNullable(broadcastState.get(table)).orElse(configMap.get(table));
                            String sinkColumns = tableProcessDim.getSinkColumns();
                            deleteNotNeedColumn(dataJsonObj, sinkColumns);
                            String type = jsonObject.getString("type");
                            dataJsonObj.put("type", type);
                            collector.collect(Tuple2.of(dataJsonObj, tableProcessDim));
                        }
                    }

                    @Override
                    public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context context, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
                        String op = tableProcessDim.getOp();
                        BroadcastState<String, TableProcessDim> broadcastState = context.getBroadcastState(mapStateDescriptor);
                        String sourceTable = tableProcessDim.getSourceTable();
                        if ("d".equals(op)) {
                            broadcastState.remove(sourceTable);
                            configMap.remove(sourceTable);
                        } else {
                            broadcastState.put(sourceTable, tableProcessDim);
                            configMap.put(sourceTable, tableProcessDim);
                        }
                    }
                }
        );
        //TODO 11.将维度数据同步到HBase对应的表中
        dimDS.print();

        dimDS.addSink(
                new RichSinkFunction<Tuple2<JSONObject, TableProcessDim>>() {

                    private Connection hBaseConn;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hBaseConn = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hBaseConn);
                    }

                    @Override
                    public void invoke(Tuple2<JSONObject, TableProcessDim> tup2, Context context) throws Exception {
                        //({"tm_name":"Redmi","id":1,"type":"update"},TableProcessDim(sourceTable=base_trademark, sinkTable=dim_base_trademark, sinkColumns=id,tm_name, sinkFamily=info, sinkRowKey=id, op=r))
                        JSONObject jsonObj = tup2.f0;
                        TableProcessDim tableProcessDim = tup2.f1;
                        //获取对业务数据库维度表操作的类型
                        String type = jsonObj.getString("type");
                        jsonObj.remove("type");

                        //获取操作的HBase表的名
                        String sinkTable = tableProcessDim.getSinkTable();
                        //获取主键
                        String rowKey = jsonObj.getString(tableProcessDim.getSinkRowKey());
                        if("delete".equals(type)){
                            //说明对业务数据库维度表进行了删除操作，需要在HBase对应的表中也执行删除
                            HBaseUtil.delRow(hBaseConn,Constant.HBASE_NAMESPACE,sinkTable,rowKey);
                        }else{
                            //说明对业务数据库维度表进行了添加或者修改，需要在HBase对应的表中执行put操作
                            String family = tableProcessDim.getSinkFamily().split(",")[0];
                            HBaseUtil.putRow(hBaseConn,Constant.HBASE_NAMESPACE,sinkTable,rowKey,family,jsonObj);
                        }
                    }
                }
        );
//        dimDS.addSink(
//                new RichSinkFunction<Tuple2<JSONObject, TableProcessDim>>() {
//                    private Connection hBaseConn;
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        hBaseConn = HBaseUtil.getHBaseConnection();
//                    }
//
//                    @Override
//                    public void close() throws Exception {
//                        HBaseUtil.closeHBaseConnection(hBaseConn);
//                    }
//
//                    @Override
//                    public void invoke(Tuple2<JSONObject, TableProcessDim> tup2, Context context) throws Exception {
//                        JSONObject jsonObj = tup2.f0;
//                        TableProcessDim tableProcessDim = tup2.f1;
//                        String type = jsonObj.getString("type");
//                        jsonObj.remove("type");
//                        String sinkTable = tableProcessDim.getSinkTable();
//                        String rowKey = jsonObj.getString(tableProcessDim.getSinkRowKey());
//                        if ("delete".equals(type)){
//                            HBaseUtil.dropHBaseTable(hBaseConn, Constant.HBASE_NAMESPACE, sinkTable);
//                        }else {
//                            String family = tableProcessDim.getSinkFamily().split(",")[0];
//                            HBaseUtil.putRow(hBaseConn, Constant.HBASE_NAMESPACE, sinkTable, rowKey, family, jsonObj);
//                        }
//
//                    }
//                }
//        );

        env.execute();

    }

    private static void deleteNotNeedColumn(JSONObject dataJsonObj, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry-> !columnList.contains(entry.getKey()));
    }
}
